package com.sc.passbook.service.impl;

import com.sc.passbook.constant.Constant;
import com.sc.passbook.mapper.FeedBackMapper;
import com.sc.passbook.service.IFeedBackService;
import com.sc.passbook.utils.RowKeyGen;
import com.sc.passbook.vo.FeedBack;
import com.sc.passbook.vo.Response;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @author sc
 * Created on  2018/9/1
 */
@Service
@Slf4j
public class FeedBackImpl implements IFeedBackService {

    @Autowired
    private HbaseTemplate hbaseTemplate;


    @Override
    public Response createFeedBack(FeedBack feedBack) {
        if(!feedBack.validate()){
            log.error("feedback 校验不通过[{}]",feedBack);
            return Response.success(feedBack);
        }
        Put put = new Put(Bytes.toBytes(RowKeyGen.genFeedBackRowKey(feedBack)));
        put.addColumn(Bytes.toBytes(Constant.FeedBackTable.FAMILY_I),
                Bytes.toBytes(Constant.FeedBackTable.USER_ID),Bytes         .toBytes(feedBack.getUserId()));

        put.addColumn(Bytes.toBytes(Constant.FeedBackTable.FAMILY_I),
                Bytes.toBytes(Constant.FeedBackTable.TEMPLATE_ID),Bytes         .toBytes(feedBack.getTemplateId()));
        put.addColumn(Bytes.toBytes(Constant.FeedBackTable.FAMILY_I),
                Bytes.toBytes(Constant.FeedBackTable.TYPE),Bytes         .toBytes(feedBack.getType()));
        put.addColumn(Bytes.toBytes(Constant.FeedBackTable.FAMILY_I),
                Bytes.toBytes(Constant.FeedBackTable.COMMENT),Bytes         .toBytes(feedBack.getComment()));
        hbaseTemplate.saveOrUpdate(Constant.FeedBackTable.TABLE_NAME,put);
        return Response.success();
    }

    @Override
    public Response getFeedBack(Long userId) {
        byte[] reverse = new StringBuilder(String.valueOf(userId)).reverse().toString().getBytes();
        Scan scan = new Scan();
        scan.setFilter(new PrefixFilter(reverse));
        List<FeedBack> list = hbaseTemplate.find(Constant.FeedBackTable.TABLE_NAME,scan,new FeedBackMapper());
        return Response.success(list);
    }
}
